package com.huitone.smspfm.task;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.huitone.smspfm.logger.ExtLogger;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.constant.DealResult;
import com.huitone.smspfm.core.filter.AttachMsgFilterTask;
import com.huitone.smspfm.core.filter.BlackListFilterTask;
import com.huitone.smspfm.core.filter.ForbidTimeFilterTask;
import com.huitone.smspfm.core.filter.IlligalUserFilterTask;
import com.huitone.smspfm.core.filter.InvalidUserFilterTask;
import com.huitone.smspfm.core.filter.RepeatMsgFilterTask;
import com.huitone.smspfm.core.filter.SendPortFilterTask;
import com.huitone.smspfm.core.filter.SensWorldFilterTask;
import com.huitone.smspfm.core.filter.SmsRouteFilterTask;
import com.huitone.smspfm.model.SmsEntityForMt;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.task.GetPriListLengthTask;

/**   
* @Title: MtSmsProcessingManager.java 
* @Package com.huitone.smspfm.module 
* @Description: 下行短信处理管理器
* @author ybliang   
* @date 2017年3月15日 下午3:51:15 
* @version V1.0   
*/
public class MtSmsProcessingTask extends Thread {
	
	private static ExtLogger logger = ExtLogger.create(MtSmsProcessingTask.class);
	
	private ExecutorService blockingPool = new ThreadPoolExecutor(AppStaticConfigs.mtsmsExcutePoolSize, 
			AppStaticConfigs.mtsmsExcutePoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.mtsmsExcutePoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private ExecutorService blockingPool2 = new ThreadPoolExecutor(AppStaticConfigs.mtsmsExcutePoolSize * 2, 
			AppStaticConfigs.mtsmsExcutePoolSize * 2, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.mtsmsExcutePoolSize * 4), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private boolean toSleep = false;
	
	private long sleepTime = 1000;
	
	private static final String CurrMtSpeedS = "CurrMtSpeedS";
	private static final String CurrMtSpeedM = "CurrMtSpeedM";
	
	private static final byte[] CURR_MT_SPEEDS_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMtSpeedS);
	private static final byte[] CURR_MT_SPEEDM_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMtSpeedM);
	
	private byte[] currMtSpeedS_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMtSpeedS);
	private byte[] currMtSpeedM_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMtSpeedM);
	
	private long currMtSpeedS = 0;
	
	private long currMtSpeedM = 0;
	
	private final Object slock = new Object();
	
	private final Object mlock = new Object();
	
	private ArrayBlockingQueue<Map<byte[],byte[]>> speeds = new ArrayBlockingQueue<Map<byte[],byte[]>>(1000);
	
	@SuppressWarnings("rawtypes")
	private DefaultRedisScript<List> script = new DefaultRedisScript<>(AppStaticConfigs.luaGetSmsCache, List.class);
	
	private List<String> keys = Collections.singletonList(AppStaticConfigs.iSMCacheList);
	
	private boolean keepGoing = true;

	@Override
	public void run() {
		
		if (AppStaticConfigs.speedStatisticsOn) {
			
			startUpdateSpeedTasks();
			
			startSpeedWriteTasks();
			
		}
		
		while(!this.isInterrupted() && isKeepGoing()) {
			
			blockingPool.execute(new Runnable() {
				
				@Override
				public void run() {
					
					try {
						
						Date sdealTime = new Date();
						
						long start = System.currentTimeMillis();

						List<List<String>> entities = AppStaticConfigs.scriptOpts.getSMCacheAndHash(script, keys, AppStaticConfigs.mtsmsBatchSizePerTime + "");
						
						if (entities == null) return;
						
						List<SmsEntityForMt> lists = new ArrayList<SmsEntityForMt>();
						for (int i = 0 ; i < entities.size() ; i ++) {
							List<String> list = entities.get(i);
							SmsEntityForMt smsEntityForMt = SmsEntityForMt.parse(list);
							lists.add(smsEntityForMt);
						}
						
						if (!lists.isEmpty()) {
							
							int waitCount = 2;
							
							if (AppStaticConfigs.illegalUserFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.invalidUserFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.blackUserFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.sensiWorldFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.dulplicateFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.forbidtimeFilterOn) {
								waitCount ++;
							}
							if (AppStaticConfigs.attachMsgFilterOn) {
								waitCount ++;
							}
							
							CyclicBarrier barrier = new CyclicBarrier(waitCount);
							
							if (AppStaticConfigs.illegalUserFilterOn) {
								new IlligalUserFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.invalidUserFilterOn) {
								new InvalidUserFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.blackUserFilterOn) {
								new BlackListFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.sensiWorldFilterOn) {
								new SensWorldFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.dulplicateFilterOn) {
								new RepeatMsgFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.forbidtimeFilterOn) {
								new ForbidTimeFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.attachMsgFilterOn) {
								new AttachMsgFilterTask(lists, barrier).start();
							}
							if (AppStaticConfigs.sendportMsgFilterOn) {
								//sendPort去掉扩展码, 过滤处理的短信,,,把扩展码设置为""  空字符串
 								new SendPortFilterTask(lists, barrier).start();
							}
						
							
							
							new SmsRouteFilterTask(lists, barrier,true).start();
							
							barrier.await();
							
							Date edealTime = new Date();
							
							blockingPool2.execute(new Runnable() {
								
								@Override
								public void run() {
									
									// 附加短信的临时数据
									List<SmsEntityForMt> attachInfos = new ArrayList<SmsEntityForMt>(); 
									// 加入重复检测队列
									List<List<String>> repeatMd5St = new ArrayList<List<String>>();
									
									// 禁发时段短信
									List<SmsEntityForMt> forbidTimeEntities = new ArrayList<>();
									
									AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											Map<String, List<byte[]>> serialnos = new HashMap<String, List<byte[]>>();
											
											for (SmsEntityForMt entity:lists) {
												
												String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
												
												Map<byte[], byte[]> map = entity.getMapB();
												
												map.put(AppCanstant.SmsEntity.SdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(sdealTime)));
												map.put(AppCanstant.SmsEntity.EdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(edealTime)));
												
												if (entity.isIntercepted()) {
													
													map.put(AppCanstant.SmsEntity.DealResult, AppStaticConfigs.stringRedisSerializer.serialize(entity.getInterceptedCode()));
													map.put(AppCanstant.SmsEntity.DealResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(entity.getInterceptedDesc()));
													
													connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
													
													connection.lPush(AppStaticConfigs.iSMInterceptListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
													
													connection.lPush(AppStaticConfigs.iSMNotifyListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum + ":2"));
													
													logger.mtwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]被过滤器拦截,处理结果:" + entity.getInterceptedDesc());
													
												} else {
													
													String priListName = entity.getPriListName();
													
													if (!StringUtils.isEmpty(priListName)) {
														// 路由成功
														if (entity.isIsforbidsms()) {
															// 属于禁发时段,则写入禁发队列
															forbidTimeEntities.add(entity);
														} else {
															// 非禁发时段,则写入优先级队列
															
															map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT0B);
															map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT0_DESCB);
															map.put(AppCanstant.SmsEntity.GateLinkId, AppStaticConfigs.stringRedisSerializer.serialize(entity.getGateLinkId()));
															
															connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
															//TODO 优先级
															if (serialnos.containsKey(priListName)) {
																serialnos.get(priListName).add(AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
															} else {
																List<byte[]> list = new ArrayList<byte[]>();
																list.add(AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
																serialnos.put(priListName, list);
															}
															
															GetPriListLengthTask.priListLength.put(priListName, StringUtils.toLong(GetPriListLengthTask.priListLength.get(priListName)) + 1);
														
															logger.mtprocc("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]路由到优先级队列" + priListName);
														}
														
														// 重复过滤数据元参数
														List<List<String>> repeatEncodes = entity.getRepeatEncodes();
														for (List<String> rh:repeatEncodes) {
															repeatMd5St.add(rh);
														}
														
														// 附加短信
														if (!StringUtils.isEmpty(entity.getAttachInfo())) {
															attachInfos.add(entity);
														}
														
													} else {
														// 路由失败
														
														map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_9B);
														map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_9_DESCB);
														
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
														
														connection.lPush(AppStaticConfigs.iSMRouteFailedListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
														
														connection.lPush(AppStaticConfigs.iSMNotifyListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum + ":2"));
														
														logger.mtwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]路由失败.");
													}
												}
											}
											
											// 以批量的方式保存serialno:number到优先级队列
											Iterator<Entry<String, List<byte[]>>> iterator = serialnos.entrySet().iterator();
											while (iterator.hasNext()) {
												
												Entry<String, List<byte[]>> entry = iterator.next();
												
												String key = entry.getKey();
												List<byte[]> value = entry.getValue();
												
												byte[] priListNameB = AppStaticConfigs.stringRedisSerializer.serialize(key);
												
												connection.lPush(priListNameB, value.toArray(new byte[1][]));
												
												connection.sAdd(AppStaticConfigs.iSMPriListNamesSetB, priListNameB);
											}
										}
									});
									
									// 禁发时段短信,保存到禁发队列
									AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											for (SmsEntityForMt entity:forbidTimeEntities) {
												
												String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
												
												Map<byte[], byte[]> map = entity.getMapB();
												
												map.put(AppCanstant.SmsEntity.SdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(sdealTime)));
												map.put(AppCanstant.SmsEntity.EdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(edealTime)));
												
												map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_1B);
												map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_1_DESCB);
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
												
												connection.zAdd(AppStaticConfigs.iSMForbidZSetB, entity.getForbidTimeEnd(), AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
												
												logger.mtwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]加入禁发时段队列.");
											}
										}
									});
									
									AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											for (SmsEntityForMt entity:attachInfos) {
												connection.zAdd(AppStaticConfigs.iSMAffixZSetB, entity.getAttachExpireTime(), AppStaticConfigs.stringRedisSerializer.serialize(entity.getAttachInfo()));
											}
											
											for (List<String> rh:repeatMd5St) {
												
												byte[] rulekey = AppStaticConfigs.stringRedisSerializer.serialize(rh.get(0));
												double timeLen = StringUtils.strToDouble(rh.get(1)) * 3600 * 1000;
												
												double score = timeLen + edealTime.getTime();
												
												connection.zAdd(AppStaticConfigs.iSMRepZSetB, score, rulekey);
											}
										}
									});
								}
							});
							
							if (AppStaticConfigs.speedStatisticsOn) {
								int size = lists.size();
								increaseCurrMtSpeedS(size);
								increaseCurrMtSpeedM(size);
							}
							
							toSleep = false;
						} else {
							toSleep = true;
						}
						
						long end = System.currentTimeMillis();
						
						long total = end - start;
						
						logger.debug("下发流程耗时" + total + "毫秒, 本次处理" + lists.size() + "条!");
						
						if (AppStaticConfigs.speedControllerOn) {
							try {
								TimeUnit.MILLISECONDS.sleep(total < 1000 ? (1000 - total) : 0);
							} catch (InterruptedException e) {}
						}
					} catch (Exception e) {
						logger.error("下行短信处理过程发生错误", e);
					}
				}
			});
			
			try {
				if (toSleep) {
					TimeUnit.MILLISECONDS.sleep(sleepTime);
				}
			} catch (InterruptedException e) {
				break;
			}
		}
		
		if (blockingPool != null && !blockingPool.isShutdown()) {
			blockingPool.shutdown();
		}
		
		if (blockingPool2 != null && !blockingPool2.isShutdown()) {
			blockingPool2.shutdown();
		}
		
		logger.info("短信平台下行短信处理服务已经关闭");
	}
	
	private void startUpdateSpeedTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					long speed = getAndResetCurrMtSpeedS();
					
					speeds.offer(Collections.singletonMap(CURR_MT_SPEEDS_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
					
					logger.debug("下行短信处理速度(条/s):" + speed);
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				int count = 0;
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					if (count >= 60) {
						
						long speed = getAndResetCurrMtSpeedM();
						
						speeds.offer(Collections.singletonMap(CURR_MT_SPEEDM_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
						
						logger.debug("下行短信处理速度(条/m):" + speed);
						
						count = 0;
						
					} else {
						count ++;
					}
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
	}
	
	private void startSpeedWriteTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while (blockingPool != null && !blockingPool.isShutdown()) {
					Map<byte[],byte[]> map = speeds.poll();
					if (map != null && !map.isEmpty()) {
						AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								connection.hMSet(AppStaticConfigs.RunningParamsHashB, map);
								// 每秒的速率队列
								byte[] bytes = map.get(CURR_MT_SPEEDS_B);
								if (bytes != null) {
									connection.lPush(currMtSpeedS_List, bytes); 
									connection.lTrim(currMtSpeedS_List, 0, AppStaticConfigs.speedSLength - 1);
								}
								// 每分的速率队列
								byte[] bytem = map.get(CURR_MT_SPEEDM_B);
								if (bytem != null) {
									connection.lPush(currMtSpeedM_List, bytem); 
									connection.lTrim(currMtSpeedM_List, 0, AppStaticConfigs.speedMLength - 1);
								}
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(1000);
						} catch (InterruptedException e) {}
					}
				}
			}
		}).start();
	}
	
	public void shutdown() {
		if (blockingPool != null) {
			blockingPool.shutdown();
		}
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
	}

	public long getCurrMtSpeedS() {
		return currMtSpeedS;
	}
	
	private long getAndResetCurrMtSpeedS() {
		long result = 0;
		synchronized (slock) {
			result = currMtSpeedS;
			currMtSpeedS = 0;
		}
		return result;
	}

	private void increaseCurrMtSpeedS(long count) {
		synchronized (slock) {
			this.currMtSpeedS += count;
		}
	}

	public long getCurrMtSpeedM() {
		return currMtSpeedM;
	}
	
	private long getAndResetCurrMtSpeedM() {
		long result = 0;
		synchronized (mlock) {
			result = currMtSpeedM;
			currMtSpeedM = 0;
		}
		return result;
	}

	private void increaseCurrMtSpeedM(long count) {
		synchronized (mlock) {
			this.currMtSpeedM += count;
		}
	}
}
